use std::sync::Arc;

use serde::Deserialize;
use tokio::sync::oneshot;

use crate::{
    resp::{Command, Value},
    RedisError, RedisResult,
};

use super::{node::NodeSocket, ConnectionState};

#[derive(Clone)]
pub(crate) struct StandaloneConnection {
    pub(crate) state: Arc<ConnectionState>,
    socket: Arc<NodeSocket>,
}

impl StandaloneConnection {
    pub fn new(host: String, port: u32, state: ConnectionState) -> Self {
        Self {
            state: Arc::new(state),
            socket: Arc::new(NodeSocket::new(host, port)),
        }
    }

    pub(crate) async fn connect(&self) -> RedisResult<()> {
        // let state = self.state.clone();
        // tokio::spawn(Self::connect_internal(state, socket));
        self.socket.set_connecting();
        let addr = format!("{}:{}", self.socket.host, self.socket.port);
        log::info!("try connect to {addr}, epoch = {}", self.socket.epcoh_get());
        super::internal_connect(&self.state, &self.socket, &addr).await
    }

    pub async fn send_command<'a, T: Deserialize<'a>>(
        &self,
        command: Command<'a>,
    ) -> RedisResult<T> {
        let bytes = command.encode();
        let rx = match self.socket.send(&bytes).await {
            Ok(rx) => rx,
            Err(err) => {
                log::error!("{err:?}");
                self.send_command_retry(&bytes).await?
            }
        };
        let val = super::wait_receive_after_send(&self.state, rx).await?;
        val.read_error()?;
        T::deserialize(&val)
    }

    async fn send_command_retry(&self, content: &[u8]) -> RedisResult<oneshot::Receiver<Value>> {
        for i in 0..self.state.retry_on_send_error {
            tokio::time::sleep(super::RETRY_INTERVAL).await;
            if i == self.state.retry_on_send_error - 1 {
                return self.socket.send(&content).await;
            } else {
                match self.socket.send(&content).await {
                    Ok(val) => {
                        return Ok(val);
                    }
                    Err(err) => {
                        log::trace!("error on retry {}, {err}", i + 1);
                    }
                }
            }
        }
        Err(RedisError::Other("error on send_command_retry".to_string()))
    }

    pub(crate) async fn send_command_no_reply<'a>(&self, command: Command<'a>) -> RedisResult<()> {
        let bytes = command.encode();
        match self.socket.send_no_reply(&bytes).await {
            Ok(_) => Ok(()),
            Err(err) => {
                log::error!("{err:?}");
                self.send_command_no_reply_retry(&bytes).await
            }
        }
    }

    async fn send_command_no_reply_retry(&self, content: &[u8]) -> RedisResult<()> {
        for i in 0..self.state.retry_on_send_error {
            tokio::time::sleep(super::RETRY_INTERVAL).await;
            if i == self.state.retry_on_send_error - 1 {
                return self.socket.send_no_reply(&content).await;
            } else {
                match self.socket.send_no_reply(&content).await {
                    Ok(_) => {
                        return Ok(());
                    }
                    Err(err) => {
                        log::trace!("error on retry no reply {}, {err}", i + 1);
                    }
                }
            }
        }
        Err(RedisError::Other(
            "error on send_command_no_reply_retry".to_string(),
        ))
    }
}
